package cn.itcast.storm;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.StormTopology;
import backtype.storm.topology.TopologyBuilder;

public class TopoMain {
	public static void main(String[] args) throws Exception{
		TopologyBuilder topologyBuilder = new TopologyBuilder();
		//设置消息源组件为RandomSpout
		topologyBuilder.setSpout("randomspout", new RandomSpout(), 4);
		
		//设置逻辑处理组件UpperBolt,并指定接收randomspout的消息
		topologyBuilder.setBolt("upperbolt", new UpperBolt(), 4).shuffleGrouping("randomspout");
		//设置逻辑处理组件SuffixBolt，并指定接收upperbolt的消息
		topologyBuilder.setBolt("suffixbolt", new SuffixBolt(), 4).shuffleGrouping("upperbolt");
		
		
		//创建一个topolopy
		StormTopology topo = topologyBuilder.createTopology();
		
		//创建一个storm的配置参数对象
		Config conf = new Config();
		//设置storm集群为这个topo启动的进程数
		conf.setNumWorkers(4);
		
		conf.setDebug(true);
		conf.setNumAckers(0);
		
		//提交topo到storm集群中
		StormSubmitter.submitTopology("demotopo", conf, topo);
	}
}
